package com.gxy.learn.thread.simple;

import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * Runnable，executor还支持另一种类型的任务——Callable。Callables也是类似于runnables的函数接口，不同之处在于，Callable返回一个值。
 * author: Gao xueyong
 * Create at: 2022/3/20 下午8:57
 */
@Slf4j
public class CallableAndFuturePool {
    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        Integer integer = afterTwoSecondsReturnVal();
        log.info("integer = {} 耗时：{}", integer, System.currentTimeMillis() - start);
        start = System.currentTimeMillis();
        Integer futureReturnVal = getFutureReturnVal();
        log.info("futureReturnVal = {} 耗时：{}", futureReturnVal, System.currentTimeMillis() - start);

        start = System.currentTimeMillis();
        List<Integer> integers = invokeAllTaskDemo();
        log.info("integers = {} 耗时：{}", integers, System.currentTimeMillis() - start);

        start = System.currentTimeMillis();
        List<String> futures = invokeAnyTaskDemo();
        log.info("futures ={} 耗时：{}",futures,System.currentTimeMillis()-start);



    }


    public static Integer afterTwoSecondsReturnVal() {
        Callable<Integer> task = () -> {
            Thread.sleep(2000L);
            return 1;
        };
        try {
            return task.call();
        } catch (Exception e) {
            log.error("err ", e);
            return null;
        }
    }

    public static Integer getFutureReturnVal() {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        Callable<Integer> task = () -> {
            Thread.sleep(5000L);
            return 1;
        };
        Future<Integer> future = executorService.submit(task);
        while (future.isDone()) {
            log.info("任务正在运行中！{}", System.currentTimeMillis());
            break;
        }
        try {
            /**
             *
             * 任何future.get()调用都会阻塞，然后等待直到callable中止。在最糟糕的情况下，一个callable持续运行——因此使你的程序将没有响应。我们可以简单的传入一个时长来避免这种情况。
             *  future.get(300,TimeUnit.SECONDS);
             */

            Integer result = future.get();
            executorService.shutdown();
            return result;
        } catch (InterruptedException e) {
            log.error("err ", e);
            return null;
        } catch (ExecutionException e) {
            log.error("err ", e);
            return null;
        }
//        catch (TimeoutException e) {
//            log.error("超时！",e);
//            return null;
//        }
    }

    /**
     * @return
     */
    public static List<Integer> invokeAllTaskDemo() {
//        newWorkStealingPool适合使用在很耗时的操作，但是newWorkStealingPool不是ThreadPoolExecutor的扩展，它是新的线程池类ForkJoinPool的扩展，但是都是在统一的一个Executors类中实现，由于能够合理的使用CPU进行对任务操作（并行操作），所以适合使用在很耗时的任务中
        ExecutorService executorService = Executors.newWorkStealingPool();
        List<Callable<Integer>> tasks = Arrays.asList(
                () -> {
                    TimeUnit.SECONDS.sleep(1);
                    return 1;
                }, () -> {
                    TimeUnit.SECONDS.sleep(2);
                    return 2;
                }, () -> {
                    TimeUnit.SECONDS.sleep(3);
                    return 3;
                });
        try {
            List<Integer> result = executorService.invokeAll(tasks).stream().map(fature -> {
                try {
                    return fature.get();
                } catch (InterruptedException e) {
                    log.error("err", e);
                } catch (ExecutionException e) {
                    log.error("err", e);
                }
                return 0;
            }).collect(Collectors.toList());
            executorService.shutdown();
            return result;
        } catch (InterruptedException e) {
            log.error("err ", e);
        }
        return null;
    }

    public static List<String> invokeAnyTaskDemo(){
        ExecutorService executorService = Executors.newWorkStealingPool(1);
        List<Callable<String>> callables = Arrays.asList(
                callable("task1", 2),
                callable("task2", 1),
                callable("task3", 3));
        try {
            return executorService.invokeAll(callables,3,TimeUnit.MINUTES).stream().map(future-> {
                try {
                    String s = future.get();
                    executorService.shutdown();
                    return s;
                } catch (InterruptedException e) {
                    log.error("err ",e);
                } catch (ExecutionException e) {
                    log.error("err ",e);
                }
                return null;
            }).collect(Collectors.toList());
        } catch (InterruptedException e) {
            log.error("err {}",e);
            return null;
        }
    }

    private static Callable<String> callable(String result, long sleepSeconds) {
        return () -> {
            TimeUnit.SECONDS.sleep(sleepSeconds);
            return result;
        };
    }
}
